package com.shujia.flink.state;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Demo7Fraud {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                //kafka broker列表
                .setBootstrapServers("master:9092,node1:9092,node2:9092")
                //指定topic
                .setTopics("transaction")
                //消费者组
                .setGroupId("Demo7Fraud")
                //读取数据的位置
                .setStartingOffsets(OffsetsInitializer.latest())
                //数据格式
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        /*
         * 001,1687160239000,100
         * 001,1687160240000,0.2
         * 001,1687160241000,100
         * 001,1687160242000,600
         * 001,1687160243000,100
         * 001,1687160247000,0.6
         * 001,1687160248000,700
         * 001,1687160249000,100
         */
        //使用kafka source
        DataStream<String> lines = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source");

        //解析数据
        DataStream<Transaction> transactions = lines.map(line -> {
            String[] split = line.split(",");
            String id = split[0];
            long tranTime = Long.parseLong(split[1]);
            double tranMoney = Double.parseDouble(split[2]);
            return new Transaction(id, tranTime, tranMoney);
        });
        /*
         * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息
         */

        //按照账户编号分组
        KeyedStream<Transaction, String> keyBys = transactions.keyBy(Transaction::getId);

        DataStream<Alert> alters = keyBys.process(new KeyedProcessFunction<String, Transaction, Alert>() {

            ValueState<Boolean> flagState;
            ValueState<Double> minTranMoneyState;
            ValueState<Long> timerState;

            @Override
            public void open(Configuration parameters) throws Exception {
                RuntimeContext context = getRuntimeContext();
                //定义一个单值状态保存标记
                flagState = context.getState(new ValueStateDescriptor<>("flag", Types.BOOLEAN));

                //定义一个状态保存小于1的金额
                minTranMoneyState = context.getState(new ValueStateDescriptor<>("minTranMoney", Types.DOUBLE));

                //用于保存定时其的时间
                timerState = context.getState(new ValueStateDescriptor<>("timer", Types.LONG));
            }

            @Override
            public void processElement(Transaction tran,
                                       KeyedProcessFunction<String, Transaction, Alert>.Context ctx,
                                       Collector<Alert> out) throws Exception {
                String id = tran.getId();
                long tranTime = tran.getTranTime();
                Double tranMoney = tran.getTranMoney();

                //从状态中获取标记
                Boolean flag = flagState.value();
                if (flag == null) {
                    flag = false;
                }

                if (flag) {
                    //紧跟着一个大于 $500 的交易
                    if (tranMoney > 500) {
                        System.out.println("紧跟着一个大于 $500 的交易");
                        //发出告警信息

                        //从状态中获取小于1的金额
                        Double minTranMoney = minTranMoneyState.value();

                        //发送数据到下游
                        out.collect(new Alert(id, tranTime, minTranMoney, tranMoney));
                    }

                    //清除状态
                    flagState.clear();
                    minTranMoneyState.clear();
                    timerState.clear();

                    //删除前一次的定时器
                    Long ts = timerState.value();
                    if (ts != null) {
                        ctx.timerService().deleteProcessingTimeTimer(ts);
                    }
                }


                //如果出现小于 $1 美元的交易
                if (tranMoney < 1) {
                    System.out.println("如果出现小于 $1 美元的交易");
                    //打上一个标记
                    flagState.update(true);
                    //将小于1的金额保存到状态中
                    minTranMoneyState.update(tranMoney);

                    //删除前一次的定时器
                    Long ts = timerState.value();
                    if (ts != null) {
                        ctx.timerService().deleteProcessingTimeTimer(ts);
                    }

                    //注册定时器，10秒之后如果没有来大于500的交易就清除状态
                    Long timestamp = ctx.timestamp();
                    ctx.timerService().registerProcessingTimeTimer(timestamp + 10000);

                    //将定时器额时间保存到状态中
                    timerState.update(timestamp + 10000);
                }
            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, Transaction, Alert>.OnTimerContext ctx, Collector<Alert> out) throws Exception {
                System.out.println("触发定时器执行");
                //清除状态
                flagState.clear();
                minTranMoneyState.clear();
            }
        });

        alters.print();

        env.execute();


    }


    @Data
    @AllArgsConstructor
    static class Transaction {
        private String id;
        private long tranTime;
        private Double tranMoney;
    }

    @Data
    @AllArgsConstructor
    static class Alert {
        private String id;
        private long tranTime;
        private Double minTranMoney;
        private Double maxTranMoney;
    }
}
